package com.atguigu.producer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

public class ConsumerDemo {
    public static void main(String[] args) throws InterruptedException {
        //1.指定配置文件
        Properties props = new Properties();
        props.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("bootstrap.servers","hadoop102:9092");
        props.setProperty("group.id","g13");
        //指定消费者第一次消费位置
        props.setProperty("auto.offset.reset","earliest");
        //允许自动提交offset
        props.setProperty("enable.auto.commit","false");

        //2.创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        ArrayList<String> topics = new ArrayList<>();
        topics.add("second");
        consumer.subscribe(topics);
        //3.消费数据

        do {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
            Iterator<ConsumerRecord<String, String>> iterator = records.iterator();
            while(iterator.hasNext()){
                ConsumerRecord<String, String> record = iterator.next();
                long offset = record.offset();
                int partition = record.partition();
                String topic = record.topic();
                String value = record.value();
                System.out.println("topic"+topic+",partition"+partition+",value"+value+",offset"+offset);
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        Set<TopicPartition> keys = offsets.keySet();
                        for (TopicPartition key : keys) {
                            OffsetAndMetadata metadata = offsets.get(key);
                            long offset1 = metadata.offset();
                            System.out.println("提交的offset是"+offset1);
                        }
                    }
                });
                //consumer.commitSync();
            }
        } while (true);


    }
}
